﻿unit TCPEngine;

interface

uses
  Winapi.Windows, Engine, Protocol, WorkThreadPool, IThread, TCPListener,
  ContextFactory, MemBuffer, WinSocket, System.SysUtils, GlobalMonitor, System.Classes;

type
  TITCPEngine = class;

  //工作引擎线程
  TITCPEngineThread = class(TIWorkThread)
  private
    _TCPEngine: TITCPEngine;
  public
    procedure CloseThread; override;
    constructor Create(FTCPEngine: TITCPEngine);
    destructor Destroy; override;
  end;
  //TCP通信引擎
  TITCPEngine = class(TIEngine)
  private
    FMaxMemPool: Integer;
    procedure SetMaxMemPool(const Value: Integer);
    //创建工作线程
    function OnCreateWorkThread: TIThread;
    //启动工作线程池
    function WorkThreadPoolStart: Boolean;
    //处理完成端口I/O消息
    procedure OnWSASocketError(FClientContext: TIClientContext;
      FBuffer: TIBuffer; bRelease: Boolean = False);
  protected
    procedure ProcessIOMessage(bRet: Boolean; dwCompletionKey: ULONG_PTR;
      pOverLappedEx: PTOverLappedEx; dwIOSize: Cardinal ); override;
  public
    constructor Create;
    function StartEngine: Boolean; override;
		procedure StopEngine; override;
  published
    property MaxMemPool: Integer read FMaxMemPool write SetMaxMemPool;
  end;

implementation

{ TITCPEngine }

constructor TITCPEngine.Create;
begin
  inherited Create;
  Started := False;
  FMaxMemPool := 2000;
end;

function TITCPEngine.OnCreateWorkThread: TIThread;
begin
  Result := TITCPEngineThread.Create(Self);
end;

procedure TITCPEngine.OnWSASocketError(FClientContext: TIClientContext;
  FBuffer: TIBuffer; bRelease: Boolean);
begin
  if FBuffer.GetOperation = IO_WSA_ACCEPTEX then
    CloseWinSocket(FBuffer.FSocket);
  MemoryPool.ReleaseBuffer(FBuffer);
  if FClientContext <> nil then
  begin
    if bRelease then
      FClientContext.ReleaseToPool
    else
      FClientContext.CloseSocket;
  end;
end;

procedure TITCPEngine.ProcessIOMessage(bRet: Boolean;
  dwCompletionKey: ULONG_PTR; pOverLappedEx: PTOverLappedEx; dwIOSize: Cardinal);
var
  dwIOError: Cardinal;
  FBuffer: TIBuffer;
  FClientContext: TIClientContext;
begin
  //得到Buffer 上下文
  FBuffer := TIBuffer(pOverLappedEx.FRequest);
  FClientContext := Pointer(dwCompletionKey);
  if FClientContext = nil then
    begin
      //仅为IO_WSA_ACCEPTEX上下文允许空
      if (pOverLappedEx.FIOType <> IO_WSA_ACCEPTEX) then
      begin
        MemoryPool.ReleaseBuffer(FBuffer);
        Exit;
      end;
    end
  else
    begin
      //GUID不同该上下文已经被占用
      if not IsEqualGUID(FClientContext.SessionID, pOverLappedEx.FGUID) then
      begin
        MemoryPool.ReleaseBuffer(FBuffer);
        Exit;
      end;
    end;
  if bRet then
    begin
      case pOverLappedEx.FIOType of
      IO_WSA_CLOSESOCKET: //一.自定义断开通知
        begin
          OnWSASocketError(FClientContext, FBuffer, True);
        end;
      IO_WSA_ACCEPTEX:
        begin
          if (not ShutDown) then
            TITCPListener(FBuffer.GetCallBackObj).CallBackAcceptEx(FBuffer)
          else
            OnWSASocketError(FClientContext, FBuffer, True);
        end;
      IO_WSA_SEND:
        begin
          FClientContext.OnWSASend(FBuffer, dwIoSize);
        end;
      IO_WSA_RECV:
        begin
          if (dwIOSize = 0) then
            OnWSASocketError(FClientContext, FBuffer)
          else
            FClientContext.OnWSARecv(FBuffer, dwIoSize);
        end;
      end;
    end
  else
    begin
      //三. 异常断开未得到正常断开的通知
      dwIOError := GetLastError();
      if dwIOError <> WAIT_TIMEOUT then
        OnWSASocketError(FClientContext, FBuffer, True);
    end;
end;

procedure TITCPEngine.SetMaxMemPool(const Value: Integer);
begin
  FMaxMemPool := Value;
end;

function TITCPEngine.StartEngine: Boolean;
begin
  Result := True;
  if not Started then
  begin
    EngineWorked := False;
//    _GlobalLogger.AppendDisplayMsg(IOCP_ENGINE_START, [FEngineName]);
//    _GlobalLogger.AppendDisplayMsg('初始化内存分配策略.', []);
    (* 启动全局性能监视器           *)
    _GlobalMonitor.StartMonitor;
    (* 内存分配                     *)
    MemoryPool.InitMemoryPool(FMaxMemPool);
    (* 启动工作线程                 *)
    if not WorkThreadPoolStart then
    begin
      Result := False;
      Exit;
    end;
//    _GlobalLogger.AppendDisplayMsg(IOCP_ENGINE_START_SUCCESS, [FEngineName]);
    ShutDown := False;
    Started := True;
  end;
end;

procedure TITCPEngine.StopEngine;
begin
  if Started then
  begin
    (* 设置引擎关闭                 *)
    ShutDown := True;
    (* 关闭工作线程                 *)
    WorkThreadPool.StopThreadPool;
    {$IFDEF _ICOP_DEBUGCLOSE}
    _GlobalLogger.AppendLogMessage('关闭所有工作线程.', []);
    {$ENDIF}
    (* 回收BUFFER内存               *)
    MemoryPool.FreeAllBuffers;
    {$IFDEF _ICOP_DEBUGCLOSE}
    _GlobalLogger.AppendLogMessage('回收所有系统内存.', []);
    {$ENDIF}
    (* 关闭IOCP性能监视器           *)
    {$IFDEF _GLOBAL_MONITOR}
    _GlobalMonitor.StopMonitor;
    {$ENDIF}
    Started := False;
  end;
end;

function TITCPEngine.WorkThreadPoolStart: Boolean;
begin
  WorkThreadPool.OnCreateThread := OnCreateWorkThread;
  Result := WorkThreadPool.StartThreadPool;
end;

{ TITCPEngineThread }

procedure TITCPEngineThread.CloseThread;
begin
  _TCPEngine.IOCompletionPort.PostIOCompletionStatus(0, 0, POverlapped(IO_WORDTHREAD_EXIT));
end;

constructor TITCPEngineThread.Create(FTCPEngine: TITCPEngine);
begin
  inherited Create(FTCPEngine.WorkThreadPool);
  _TCPEngine := FTCPEngine;
end;

destructor TITCPEngineThread.Destroy;
begin
  inherited Destroy;
end;

end.
